package com.to8to.kitt.esb;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.curator.utils.ThreadUtils;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageMeta;
import org.apache.thrift.protocol.TMessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.Constants;
import com.to8to.kitt.ThriftMessage;
import com.to8to.kitt.ThriftMessageCodec;
import com.to8to.kitt.esb.http.HttpBusServer;
import com.to8to.kitt.utils.DateUtil;
import com.to8to.kitt.utils.LogPatterns;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;
import com.to8to.kitt.utils.UniqueIdGenerator;

@Sharable
public class BusChannelHandler extends ChannelHandlerAdapter {
	
	public static Logger logger = LoggerFactory.getLogger(BusChannelHandler.class);  

	private Map<String, BusMessageState> stateMap = new HashMap<String, BusMessageState>();
	
	private ReadWriteLock lock = new ReentrantReadWriteLock();     
	
	private ZKServiceChannelPool serviceChannelPool;
	
	public ExecutorService executor = ThreadUtils.newSingleThreadExecutor("esb-clean");// Executors.newSingleThreadExecutor();
	
	private void putState(String key, BusMessageState state)
	{
		lock.writeLock().lock();
		
		try
		{
			stateMap.put(key, state);
		}
		catch(Exception ee)
		{
			
		}
		finally
		{
			lock.writeLock().unlock();
		}
	}
	
	
	private BusMessageState getAndRemoveState(String key)
	{
		lock.writeLock().lock();
		try
		{
			BusMessageState v = stateMap.get(key);
			
			stateMap.remove(key);
			
			return v;
		}
		catch(Exception ee)
		{
			
		}
		finally
		{
			lock.writeLock().unlock();
		}
		return null;
	}
	
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    	logger.info("链路注册:{}", ctx.channel());
    	super.channelRegistered(ctx);
    }
	
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    public ZKServiceChannelPool getServiceChannelPool() {
		return serviceChannelPool;
	}

	public void setServiceChannelPool(ZKServiceChannelPool serviceChannelPool) {
		this.serviceChannelPool = serviceChannelPool;
	}
	
	public BusChannelHandler()
	{
		launchCleanDeamon();
	}

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) 
    {
    	if(msg instanceof BusMessage)
    	{
    		try {
				handle(ctx, (BusMessage)msg);
	        } 
	    	catch(Throwable e)
	    	{
	    		NettyUtils.exitIfOutOfMemory(e);
	    		logger.error(LogUtils.logStackTrace(e));
	    		e.printStackTrace();
	    	}
	    	finally {
	        	ReferenceCountUtil.release(msg);
	        }
    	}
    }
    
    
	public void launchCleanDeamon()
	{
		executor.execute(new Runnable(){
			@Override
			public void run() {
				
				while(true)
				{
					long now = System.currentTimeMillis();
					
					lock.writeLock().lock();
					
					try
					{
						Set<String> keyset = stateMap.keySet();
						for(String key: keyset)
						{
							BusMessageState status = stateMap.get(key);
							if(status == null)
							{
								stateMap.remove(key);
							}
							else
							{
								if(now - status.getRecvFromClientTime() > Constants.ESB_EXP_TIME_MS) //timeout
								{
									logger.trace("清理超时的任务: " + status.getName() + ", 收到请求时间: " + DateUtil.format(new Date(status.getRecvFromClientTime())));
									
									if(status.inMsg != null && status.inMsg.getBuff() != null && status.inMsg.getBuff().refCnt() > 0)
										status.inMsg.getBuff().release();
									if(status.outMsg != null && status.outMsg.getBuff() != null && status.outMsg.getBuff().refCnt() > 0)
										status.outMsg.getBuff().release();
									status = null;
									
									stateMap.remove(key);
								}
							}
						}
					}
					catch(Exception ee)
					{
						logger.error(LogUtils.logStackTrace(ee));
					}
					finally
					{
						lock.writeLock().unlock();
					}
					
					
					HttpBusServer.cleanOnce();
					
					try {
						Thread.sleep(Constants.ESB_EXP_TIME_MS);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
		});
	}
    
    public static String genKey(String name, int seqid)
    {
    	return name + "|" + seqid;
    }
    
    
	public void handle(ChannelHandlerContext ctx, BusMessage inMsg) 
	{	
		long t1 = System.currentTimeMillis();
		
		TMessage head = inMsg.getHead();
		
		TMessageMeta meta = inMsg.getMeta();
		
		BusMessageState status = null;
		
		boolean isError = false;
		
		int errorType = 0;
		
		String errorMsg = "";
		
		LogUtils.openTag(meta.tag);
		
		if(inMsg.isCall()) 
    	{
//			if(meta.service.equals(BusStatusService.class.getName()))
//			{
//				statusManager.handleStatQuery(ctx, inMsg);
//				LogUtils.removeTag();
//				return ;
//			}
			
			int oldId = head.seqid;
			
			int newId = UniqueIdGenerator.uniqueInt();
			
			String msgKey = genKey(head.name, newId);
			
			while(stateMap.containsKey(msgKey))
			{
				logger.warn(LogPatterns.FAIL_TAG + " msgKey重复: {}, 重新生成  ... ", msgKey);
				newId = UniqueIdGenerator.uniqueInt(ctx.channel().remoteAddress().toString(), head.name);
				msgKey = genKey(head.name, newId);
			}
			
			logger.debug(LogPatterns.RECV_FROM_CLIENT, ctx.channel().toString(), ctx.channel().remoteAddress().toString(), oldId, newId, head.name);
			
			inMsg.setSeqid(newId);
			
			inMsg.getBuff().readerIndex(0);
			
			status = new BusMessageState(msgKey, oldId, newId, meta, ctx, inMsg);
			
			status.setRecvFromClientTime(t1);
			
			Entry<String, Channel> entry = serviceChannelPool.getChannelMT(meta.service);
			
			if(entry != null)
			{
				logger.debug("路由成功，找到下游服务:{}, 地址:{}, 转发请求", meta.service, entry.toString());
				
				status.setDownpool(entry.getValue());
				
				ChannelFutureListener l = new SDFutureListener(entry, status, serviceChannelPool);
				
				status.recordStartSendDownTime();
				
				putState(status.getKey(), status);
				
	            ChannelFuture cf = entry.getValue().writeAndFlush(inMsg);
	            
	            cf.addListener(l);
			}
			else 
			{
				isError = true;
				errorType = TApplicationException.INTERNAL_ERROR;
				errorMsg = LogPatterns.FAIL_TAG + " 下游服务路由失败，或者跟该主机的连接失败!!! 消息头: " + inMsg.getHead();
			}
    	}
		else if(inMsg.isReply() || inMsg.isException())
		{
			int newId = inMsg.getSeqid();
			
			String msgKey = genKey(head.name, newId);

			logger.debug(LogPatterns.RECV_FROM_DOWN, ctx.channel().toString(), ctx.channel().remoteAddress().toString(), "", newId, inMsg.getHead().name);

			status = getAndRemoveState(msgKey);
			
			if(status != null)
			{
				status.outMsg = inMsg;
				
				status.setRecvFromDownTime(t1);
				
				ChannelHandlerContext ctxUp = status.getUpctx();
				
				if(ctxUp != null)
				{
					logger.debug("找回客户端链路: {} \n<=== 开始发送客户端，返回消息头: {}", ctxUp.channel(), inMsg.getHead());
					
					inMsg.setSeqid(status.getOldseq());
					
					status.recordStartSendClientTime();
					
					ChannelFutureListener l = new SCFutureListener(status);
					
					ChannelFuture cf = ctxUp.writeAndFlush(inMsg);

					cf.addListener(l);
				}
				else 
				{
					logger.error(LogPatterns.FAIL_TAG + " 客户端链路为空: {}, 消息key:{} ", status.getUpctx(), status.getKey());
				}
			}
			else 
			{
				isError = true;
				errorType = TApplicationException.BAD_SEQUENCE_ID;
				errorMsg = LogPatterns.FAIL_TAG + " 下游回复的消息状态找不到!!! 消息头: " + inMsg.getHead() + ", msgKey: " + msgKey;
			}
		}
		else
		{
			isError = true;
			errorType = TApplicationException.INVALID_MESSAGE_TYPE;
			errorMsg = LogPatterns.FAIL_TAG + " 非法消息类型!!! 消息头: " + inMsg.getHead();
		}
		

		if(isError)
		{
			TMessage h = new TMessage(inMsg.getHead().name, TMessageType.EXCEPTION, inMsg.getSeqid());
			ThriftMessage outMsg = new ThriftMessage(h, new TApplicationException(errorType, errorMsg));
			outMsg.setMeta(meta);
			logger.error(errorMsg);
			if(status != null)
			{
				ByteBuf buf = ThriftMessageCodec.encode(outMsg);
				
				BusMessage bmsg = new BusMessage();
				bmsg.setHead(h);
				bmsg.setBuff(buf);
				bmsg.setMeta(meta);
				
				if(status.getUpctx() != null)
				{
					inMsg.getBuff().release();
					
					ChannelFutureListener l = new SCFutureListener(status);
					
					logger.debug("<=== 开始发送失败回复 , 链路: {}", status.getUpctx().channel());
					
					ChannelFuture cf = status.getUpctx().writeAndFlush(bmsg);
					
					cf.addListener(l);
				}
				else 
				{
					logger.error(LogPatterns.FAIL_TAG + " 发送异常回复失败!");
				}
			}
			else
			{
				logger.error(LogPatterns.FAIL_TAG + " 发送异常回复失败!");
			}
		}
		
		LogUtils.removeTag();
	}
		
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    	NettyUtils.exitIfOutOfMemory(cause);
    	logger.trace(
    			"\n####################################################################################\n"
    			+ "##############  exceptionCaught: \n" + LogUtils.logStackTrace(cause) + "\n" 
    			+ "##############  Channel: \n" + ctx.channel() + "\n"
    			+ "####################################################################################"
    			);
    }
}
